package git.soulbgm.hbase;

import git.soulbgm.hbase.annotation.HBaseColumn;
import git.soulbgm.hbase.annotation.HBaseTable;
import git.soulbgm.utils.ClassUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.assertj.core.util.Lists;

import java.io.IOException;
import java.lang.reflect.Field;
import java.util.*;

/**
 * HBase工具类
 *
 * @author SoulBGM
 * @version 1.0
 * @date 2020/6/18 10:22
 */
@Slf4j
public class HBaseUtil {

    /**
     * 创建HBase表
     *
     * @param tableName    表名称
     * @param familyColumn 多个列族
     * @return 成功true 失败false
     */
    public boolean createTable(String tableName, String... familyColumn) {
        return createTable(tableName, new HashSet<>(Arrays.asList(familyColumn)));
    }

    /**
     * 创建HBase表
     *
     * @param tableName       表名称
     * @param familyColumnSet 多个列族Set集合
     * @return 成功true 失败false
     */
    public boolean createTable(String tableName, Set<String> familyColumnSet) {
        //设置表名
        TableName tn = TableName.valueOf(tableName);
        //将表名传递给HTableDescriptor
        HTableDescriptor htd = new HTableDescriptor(tn);
        for (String family : familyColumnSet) {
            //创建列族
            HColumnDescriptor hcd = new HColumnDescriptor(family);
            //将列族添加进表中
            htd.addFamily(hcd);
        }
        //Admin是操作表的类，具有创建
        try (Admin admin = HBaseConnectionFactory.getConnection().getAdmin()) {
            if (admin.tableExists(htd.getTableName())) {
                log.warn("HBase中已经存在{}表,请修改名称或删除该表后重建", tableName);
                return false;
            }
            // 创建表
            admin.createTable(htd);
            return true;
        } catch (IOException e) {
            log.warn("创建HBase的{}表失败", tableName, e);
            return false;
        }
    }

    /**
     * 删除HBase表
     *
     * @param tableName 表名称
     * @return 成功true 失败false
     */
    public boolean dropTable(String tableName) {
        TableName tn = TableName.valueOf(tableName);
        try (Admin admin = HBaseConnectionFactory.getConnection().getAdmin()) {
            // 禁用表
            admin.disableTable(tn);
            // 删除表
            admin.deleteTable(tn);
            return true;
        } catch (IOException e) {
            log.warn("删除HBase的{}表失败", tableName, e);
            return false;
        }
    }

    /**
     * 保存数据
     *
     * @param tableName 表名称
     * @param putList   需要保存的数据
     * @return 成功true 失败false
     */
    public boolean savePut(String tableName, List<Put> putList) {
        if (isBlank(tableName)) {
            return false;
        }
        try (Table table = HBaseConnectionFactory.getConnection().getTable(TableName.valueOf(tableName))) {
            table.put(putList);
            return true;
        } catch (IOException e) {
            log.warn("数据存入HBase的{}表失败", tableName, e);
            return false;
        }
    }

    /**
     * 通过实体类获得表名
     *
     * @param obj 实体类
     * @return 表名
     */
    public String getOrmTable(Object obj) {
        HBaseTable hbaseTable = ClassUtil.getBeanAnnotation(obj, HBaseTable.class);
        if (hbaseTable == null) {
            return null;
        } else {
            if ("".equals(hbaseTable.value())) {
                return ClassUtil.camelToUnderline(obj.getClass().getSimpleName());
            }
            return hbaseTable.value();
        }
    }

    /**
     * 保存数据到HBase 并检查表是否存在
     *
     * @param dataList 数据列表
     * @return 成功true 失败false
     */
    public <T> boolean saveCheck(List<T> dataList) {
        if (dataList == null) {
            return false;
        }
        if (dataList.isEmpty()) {
            return true;
        }
        T obj = dataList.get(0);
        String tableName = getOrmTable(obj);
        try (Admin admin = HBaseConnectionFactory.getConnection().getAdmin()) {
            if (!admin.isTableAvailable(TableName.valueOf(tableName))) {
                // 表不存在，先获取family创建表
                Class<?> clazz = obj.getClass();
                Field[] fields = clazz.getDeclaredFields();
                Set<String> familyColumnSet = new HashSet<>(2);
                for (Field field : fields) {
                    if (!ClassUtil.existFieldAnnotation(field, HBaseColumn.class)) {
                        continue;
                    }
                    HBaseColumn column = ClassUtil.getFieldAnnotation(field, HBaseColumn.class);
                    String family = column.family();
                    familyColumnSet.add(family);
                }
                // 创建表
                createTable(tableName, familyColumnSet);
            }
            return save(tableName, dataList);
        } catch (IOException e) {
            log.warn("保存数据到{}表出现异常", tableName, e);
        }
        return false;
    }

    /**
     * 保存数据到HBase
     *
     * @param dataList 数据列表
     * @return 成功true 失败false
     */
    public <T> boolean save(List<T> dataList) {
        if (dataList == null) {
            return false;
        }
        if (dataList.isEmpty()) {
            return true;
        }
        String tableName = getOrmTable(dataList.get(0));
        return save(tableName, dataList);
    }

    /**
     * 保存数据到HBase
     *
     * @param dataList  数据列表
     * @param tableName 表名
     * @return 成功true 失败false
     */
    public <T> boolean save(String tableName, List<T> dataList) {
        List<Put> putList = new ArrayList<>(dataList.size());
        for (T t : dataList) {
            if (t == null) {
                continue;
            }
            Put put = HBaseBeanUtil.entityToPut(t);
            if (put != null) {
                putList.add(put);
            }
        }
        return savePut(tableName, putList);
    }

    /**
     * 批量删除数据
     *
     * @param tableName 表名称
     * @param rowkeys   要删除的rowkey
     * @return 成功true 失败false
     */
    public boolean delete(String tableName, String... rowkeys) {
        if (isBlank(tableName)) {
            return false;
        }
        if (rowkeys == null || rowkeys.length == 0) {
            return true;
        }
        List<Delete> deleteList = new ArrayList<>(rowkeys.length);
        for (String rowkey : rowkeys) {
            if (isBlank(rowkey)) {
                continue;
            }
            deleteList.add(new Delete(Bytes.toBytes(rowkey)));
        }
        return delete(tableName, deleteList);
    }

    /**
     * 批量删除数据
     *
     * @param tableName  表名称
     * @param deleteList 要删除的数据
     * @return 成功true 失败false
     */
    public boolean delete(String tableName, List<Delete> deleteList) {
        if (isBlank(tableName)) {
            return false;
        }
        try (Table table = HBaseConnectionFactory.getConnection().getTable(TableName.valueOf(tableName))) {
            table.delete(deleteList);
            return true;
        } catch (IOException e) {
            log.warn("删除数据失败", e);
            return false;
        }
    }

    /**
     * 通过rowKey获得HBase表数据
     *
     * @param tableName 表名称
     * @param rowKeys   rowKey
     * @return 成功true 失败false
     */
    public List<Result> getResult(String tableName, String... rowKeys) {
        try (Table table = HBaseConnectionFactory.getConnection().getTable(TableName.valueOf(tableName))) {
            return getResult(table, rowKeys);
        } catch (IOException e) {
            log.warn("获取HBase的{}表数据失败", tableName, e);
            return Lists.newArrayList();
        }
    }

    /**
     * 通过rowKey获得HBase表数据
     *
     * @param table   HBase表
     * @param rowKeys rowKey
     * @return 成功true 失败false
     */
    public List<Result> getResult(Table table, String... rowKeys) {
        List<Result> resultList = new ArrayList<>();
        List<Get> getList = new ArrayList<>();
        for (String r : rowKeys) {
            if (isBlank(r)) {
                continue;
            }
            Get get = new Get(Bytes.toBytes(r));
            getList.add(get);
        }
        try {
            Result[] results = table.get(getList);
            Collections.addAll(resultList, results);
        } catch (IOException e) {
            log.warn("获取HBase的{}表数据失败", table.getName().getNameAsString(), e);
        }
        return resultList;
    }

    /**
     * 通过rowkey查询HBase表
     * *
     *
     * @param tableName 表名称
     * @param clazz     反射的实体Class
     * @param rowKeys   rowKey
     * @return HBase表数据
     */
    public <T> List<T> get(String tableName, Class<T> clazz, String... rowKeys) {
        List<T> dataList = new ArrayList<>();
        try (
                Table table = HBaseConnectionFactory.getConnection().getTable(TableName.valueOf(tableName));
                Admin admin = HBaseConnectionFactory.getConnection().getAdmin()
        ) {
            if (!admin.isTableAvailable(TableName.valueOf(tableName))) {
                return dataList;
            }
            List<Result> resultList = getResult(table, rowKeys);
            if (resultList.isEmpty()) {
                return dataList;
            }
            for (Result result : resultList) {
                if (result == null || result.isEmpty()) {
                    continue;
                }
                T data = HBaseBeanUtil.resultToEntity(result, clazz);
                dataList.add(data);
            }
        } catch (IOException e) {
            log.warn("数据存入HBase的{}表失败", tableName, e);
        }
        return dataList;
    }

    /**
     * 通过Scan进行查询
     *
     * @param tableName 表名
     * @param clazz     反射的实体Class
     * @param scan      Scan类
     * @return HBase表数据
     */
    public <T> List<T> scan(String tableName, Class<T> clazz, Scan scan) {
        List<T> dataList = new ArrayList<>();
        try (
                Table table = HBaseConnectionFactory.getConnection().getTable(TableName.valueOf(tableName));
                Admin admin = HBaseConnectionFactory.getConnection().getAdmin()
        ) {
            if (!admin.isTableAvailable(TableName.valueOf(tableName))) {
                return dataList;
            }
            ResultScanner scanner = table.getScanner(scan);
            Iterator<Result> iterator = scanner.iterator();
            List<Result> resultList = new ArrayList<>();
            while (iterator.hasNext()) {
                resultList.add(iterator.next());
            }
            if (resultList.isEmpty()) {
                return dataList;
            }
            for (Result result : resultList) {
                if (result == null || result.isEmpty()) {
                    continue;
                }
                T data = HBaseBeanUtil.resultToEntity(result, clazz);
                dataList.add(data);
            }
        } catch (IOException e) {
            log.warn("数据存入HBase的{}表失败", tableName, e);
        }
        return dataList;
    }

    public boolean isEmpty(String str) {
        return str == null || str.length() == 0;
    }

    public boolean isBlank(String str) {
        int strLen;
        if (str == null || (strLen = str.length()) == 0) {
            return true;
        }
        for (int i = 0; i < strLen; i++) {
            if (!Character.isWhitespace(str.charAt(i))) {
                return false;
            }
        }
        return true;
    }

}
